use crate::cmp::{Cmp, InternalKeyCmp};
use crate::error::Result;
use crate::key_types::{parse_internal_key, InternalKey, LookupKey, UserKey, ValueType};
use crate::table_cache::TableCache;
use crate::table_reader::TableIterator;
use crate::types::{FileMetaData, FileNum, LdbIterator, Shared, MAX_SEQUENCE_NUMBER, NUM_LEVELS};

use bytes::Bytes;
use std::cmp::Ordering;
use std::default::Default;
use std::rc::Rc;

/// FileMetaHandle is a reference-counted FileMetaData object with interior mutability. This is
/// necessary to provide a shared metadata container that can be modified while referenced by e.g.
/// multiple versions.
pub type FileMetaHandle = Shared<FileMetaData>;

/// Contains statistics about seeks occurred in a file.
/// GetStats 结构体用于记录数据库中一次查询操作（get 操作）的统计信息，主要追踪查询命中的具体文件及其所在层级，帮助分析查询性能或调试问题。
pub struct GetStats {
    /// 存储查询命中的 SSTable 文件的元数据句柄（FileMetaHandle）。
    /// Some(handle) 表示查询在该文件中找到了目标键；
    /// None 表示未命中任何文件（目标键不存在或查询失败）。
    /// 元数据句柄包含文件编号、大小、键范围等信息，可用于定位具体的物理文件。
    file: Option<FileMetaHandle>,
    /// 记录命中文件所在的层级（Level 0 到 Level N-1）。
    /// 不同层级的文件具有不同的特性（如 Level 0 可能有重叠键范围，Level 1+ 键范围不重叠），记录层级有助于分析查询路径（例如，是否频繁命中高层级文件，可能需要优化压缩策略）。
    level: usize,
}

/// Version 结构体是数据库某一时刻的版本快照，包含了该版本下所有 SSTable 文件的布局、压缩相关状态等核心元数据。
/// 它是存储引擎实现多版本管理和高效查询的基础。各个字段的含义如下：
pub struct Version {
    /// 指向 SSTable 文件的缓存（TableCache），用于快速访问磁盘上的 SSTable 数据块（避免频繁打开 / 关闭文件）。Shared 表示可能被多个版本或线程共享，提高缓存利用率。
    table_cache: Shared<TableCache>,
    /// 用户定义的键比较器（实现 Cmp 特性），用于键的排序和范围比较。Rc 包装表示可能被多个版本共享，确保同一数据库实例中比较逻辑一致。
    user_cmp: Rc<Box<dyn Cmp>>,
    /// 数据库所有层级（Level 0 到 Level N-1）的文件列表数组，是 Version 最核心的字段：
    /// 数组索引对应层级（files[0] 是 Level 0 的文件，files[1] 是 Level 1 的文件，以此类推）。
    /// 每个元素是 Vec<FileMetaHandle>，存储该层级所有 SSTable 文件的元数据句柄（FileMetaHandle 包含文件编号、大小、键范围等信息）。
    /// 这些文件共同构成了该版本的完整数据视图，查询时会按层级依次查找。
    pub files: [Vec<FileMetaHandle>; NUM_LEVELS],

    /// 标记当前需要压缩的单个文件（可选）。例如，当某个文件的查询次数过多（超过 allowed_seeks 阈值）时，会被标记为需要压缩，以优化后续查询性能。
    pub file_to_compact: Option<FileMetaHandle>,
    /// 与 file_to_compact 配合，记录需要压缩的文件所在的层级。
    pub file_to_compact_lvl: usize,
    /// 该版本的最高 “压缩分数”（由 finalize 方法计算），用于判断是否需要触发压缩：
    /// 当分数 ≥1.0 时，通常会触发对应层级的压缩。
    /// 分数越高，说明该层级的数据布局越 “拥挤”（文件过多或总大小超标）。
    pub compaction_score: Option<f64>,
    /// 记录需要进行压缩的层级（由 finalize 方法确定），即 compaction_score 最高的层级。后台压缩线程会优先处理该层级。
    pub compaction_level: Option<usize>,
}

struct DoSearchResult(Option<(Vec<u8>, Vec<u8>)>, Vec<FileMetaHandle>);

impl Version {
    pub fn new(cache: Shared<TableCache>, ucmp: Rc<Box<dyn Cmp>>) -> Version {
        Version {
            table_cache: cache,
            user_cmp: ucmp,
            files: Default::default(),
            file_to_compact: None,
            file_to_compact_lvl: 0,
            compaction_score: None,
            compaction_level: None,
        }
    }

    pub fn num_level_bytes(&self, l: usize) -> usize {
        assert!(l < NUM_LEVELS);
        total_size(self.files[l].iter())
    }

    pub fn num_level_files(&self, l: usize) -> usize {
        assert!(l < NUM_LEVELS);
        self.files[l].len()
    }

    /// get returns the value for the specified key using the persistent tables contained in this
    /// Version.
    ///
    /// 这个 get 方法是数据库的核心查询接口，用于根据传入的内部键（InternalKey）查找对应的值，是读取数据的主要入口。
    ///
    /// 它会在整个 LSM 树的层级结构中搜索目标键，并返回找到的最新有效值（或确认键已被删除）。
    #[allow(unused_assignments)]
    pub fn get(&self, key: InternalKey<'_>) -> Result<Option<(Bytes, GetStats)>> {
        // 获取相关层级文件：self.get_overlapping(key) 找到所有与目标键（key）的键范围重叠的 SSTable 文件，按层级组织（从 Level 0 到更高层级）。
        let levels = self.get_overlapping(key);
        let ikey = key;
        // 解析用户键：通过 parse_internal_key(ikey).2 从内部键中提取用户实际查询的原始键（ukey）。
        let ukey = parse_internal_key(ikey).2;
        // 初始化统计信息：GetStats 用于记录查询过程中访问的文件和层级，便于监控和优化。
        let mut stats = GetStats {
            file: None,
            level: 0,
        };

        // 循环遍历每个层级（level）及其包含的重叠文件（files），按层级从低到高（Level 0 优先）搜索：
        for (level, files) in levels.iter().enumerate() {
            // 记录访问信息
            let mut last_read = None;
            let mut last_read_level: usize = 0;
            for f in files {
                if last_read.is_some() && stats.file.is_none() {
                    stats.file.clone_from(&last_read);
                    stats.level = last_read_level;
                }
                last_read_level = level;
                last_read = Some(f.clone());

                // We receive both key and value from the table. Because we're using InternalKey
                // keys, we now need to check whether the found entry's user key is equal to the
                // one we're looking for (get() just returns the next-bigger key).

                //  self.table_cache.borrow_mut().get(f.borrow().num, ikey) 这里 BlockIter 进
                if let Ok(Some((k, v))) = self.table_cache.borrow_mut().get(f.borrow().num, ikey) {
                    // We don't need to check the sequence number; get() will not return an entry
                    // with a higher sequence number than the one in the supplied key.
                    let (typ, _, foundkey) = parse_internal_key(&k);
                    if typ == ValueType::TypeValue && self.user_cmp.cmp(foundkey, ukey) == Ordering::Equal {
                        return Ok(Some((v, stats)));
                    } else if typ == ValueType::TypeDeletion {
                        // Skip looking once we have found a deletion.
                        return Ok(None);
                    }
                }
            }
        }
        Ok(None)
    }

    /// get_overlapping returns the files overlapping key in each level.
    ///
    /// 这个 get_overlapping 方法用于查找 LSM 树中所有与目标键（key）的键范围重叠的 SSTable 文件，
    ///
    /// 并按层级（Level 0 到最高层级）组织这些文件，为后续的查询操作（如 get 方法）提供需要扫描的文件列表。
    fn get_overlapping(&self, key: InternalKey<'_>) -> [Vec<FileMetaHandle>; NUM_LEVELS] {
        // 创建一个长度为 NUM_LEVELS（LSM 树的总层级数）的数组 levels，用于存储每个层级中与目标键重叠的文件。
        let mut levels: [Vec<FileMetaHandle>; NUM_LEVELS] = Default::default();
        // 从目标内部键（key）中解析出用户原始键（ukey），后续用于判断键范围重叠。
        let ikey = key;
        let ukey = parse_internal_key(key).2;

        let files = &self.files[0];
        levels[0].reserve(files.len());
        for f_ in files {
            let f = f_.borrow();
            // 解析文件的最小键和最大键（用户键部分）
            let (fsmallest, flargest) = (
                parse_internal_key(&f.smallest).2,
                parse_internal_key(&f.largest).2,
            );
            // 判断目标用户键是否在 [fsmallest, flargest] 范围内
            if self.user_cmp.cmp(ukey, fsmallest) >= Ordering::Equal && self.user_cmp.cmp(ukey, flargest) <= Ordering::Equal {
                levels[0].push(f_.clone()); // 加入重叠文件列表
            }
        }
        // Sort by newest first.
        // Level 0 特点：文件之间允许键范围重叠，因此需要检查所有文件是否与目标键重叠。
        // 按文件编号降序排序（编号越大，文件越新）
        levels[0].sort_by(|a, b| b.borrow().num.cmp(&a.borrow().num));

        let icmp = InternalKeyCmp(self.user_cmp.clone());
        (1..NUM_LEVELS).for_each(|level| {
            let files = &self.files[level];
            // 查找可能与目标键重叠的文件（利用高层级文件不重叠且有序的特性）
            if let Some(ix) = find_file(&icmp, files, ikey) {
                let f = files[ix].borrow();
                let fsmallest = parse_internal_key(&f.smallest).2;
                // 再次验证目标键是否在文件的键范围内
                if self.user_cmp.cmp(ukey, fsmallest) >= Ordering::Equal {
                    levels[level].push(files[ix].clone());
                }
            }
        });

        levels
    }

    /// level_summary returns a summary of the distribution of tables and bytes in this version.
    ///
    /// 这个 level_summary 方法用于生成当前数据库版本（Version）中各层级 SSTable 文件的分布摘要，以人类可读的字符串形式呈现每个层级的文件数量、总字节数及具体文件信息，是数据库监控、调试和状态查看的核心工具。
    ///
    pub fn level_summary(&self) -> String {
        let mut acc = String::with_capacity(256);  // 预分配 256 字节容量，优化性能
        for level in 0..NUM_LEVELS {  // 遍历所有层级（从 Level 0 到 Level N-1）
            // self.files[level]：存储当前版本下第 level 层级的所有 SSTable 文件元数据（FileMetaHandle）；
            // total_size 辅助函数：遍历文件集合，累加所有文件的 size 字段，得到层级总字节数；
            // 格式化规则：每个层级的信息包含 “层级编号、文件数、总字节数、具体文件列表”，结构清晰且便于调试。
            let fs = &self.files[level]; // 获取当前层级的所有 SSTable 文件
            if fs.is_empty() {
                continue;  // 跳过无文件的层级，避免冗余输出
            }
            // 1. 提取当前层级所有文件的关键信息（文件编号 + 文件大小）
            let filedesc: Vec<(FileNum, usize)> = fs
                .iter()
                .map(|f| (f.borrow().num, f.borrow().size))  // 每个文件的 (编号, 字节数)
                .collect();
            // 2. 格式化当前层级的摘要信息
            let desc = format!(
                "level {}: {} files, {} bytes ({:?}); ",
                level,  // 层级编号
                fs.len(), // 该层级的文件总数
                total_size(fs.iter()),  // 该层级所有文件的总字节数（通过辅助函数计算）
                filedesc  // 该层级所有文件的总字节数（通过辅助函数计算）
            );
            // 3. 将当前层级摘要追加到总摘要中
            acc.push_str(&desc);
        }
        acc
    }

    /// 这个 pick_memtable_output_level 方法用于决定内存表（MemTable）的数据在持久化时，应该直接写入到 LSM 树的哪一层级（Level），而非固定写入 Level 0。
    /// 其核心目标是优化文件布局，减少后续压缩操作的开销。
    pub fn pick_memtable_output_level(&self, min: UserKey<'_>, max: UserKey<'_>) -> usize {
        //  初始默认层级：Level 0
        let mut level = 0;
        // 第一层判断：Level 0 是否有键范围重叠
        if !self.overlap_in_level(0, min, max) {
            // Go to next level as long as there is no overlap in that level and a limited overlap
            // in the next-higher level.
            // // 若 Level 0 与当前数据的键范围（min~max）无重叠，才尝试提升层级
            let start = LookupKey::new(min, MAX_SEQUENCE_NUMBER);
            let limit = LookupKey::new_full(max, 0, ValueType::TypeDeletion);

            const MAX_MEM_COMPACT_LEVEL: usize = 2; // 最多提升到 Level 2
            while level < MAX_MEM_COMPACT_LEVEL {
                // 条件1：下一层级（level+1）是否与当前键范围无重叠
                if self.overlap_in_level(level + 1, min, max) {
                    break;
                }
                // 条件2：再下一层级（level+2）的重叠文件大小是否可控
                if level + 2 < NUM_LEVELS {
                    let overlaps = self.overlapping_inputs(
                        level + 2,
                        start.internal_key(),
                        limit.internal_key(),
                    ); // 获取 level+2 中重叠的文件
                    let size = total_size(overlaps.iter());  // 计算重叠文件总大小
                    if size > 10 * (2 << 20) { // 若超过 20MB（10*(2^21) 字节）
                        break; // 重叠文件过大，停止提升
                    }
                }
                level += 1; // 满足所有条件，提升到下一层级
            }
        }
        level
    }

    /// record_read_sample returns true if there is a new file to be compacted. It counts the
    /// number of files overlapping a key, and which level contains the first overlap.
    ///
    /// 这段代码定义了 record_read_sample 方法，主要用于在存储引擎（如 LSM 树）中记录读取操作的样本信息，并判断是否需要触发新的文件压缩（compaction）。
    #[allow(unused_assignments)]
    pub fn record_read_sample(&mut self, key: InternalKey<'_>) -> bool {
        let mut contained_in = 0;
        let mut i = 0;
        let mut first_file = None;
        let mut first_file_level = None;
        // 通过 self.get_overlapping(key) 获取所有与目标键（key）重叠的 SSTable 文件，按层级（level）组织
        self.get_overlapping(key).iter().for_each(|level| {
            // 遍历这些层级，统计：
            // contained_in：与该键重叠的文件总数（多个层级中可能存在包含同一键范围的文件）
            // first_file 和 first_file_level：第一个包含该键的文件及其所在层级（用于后续统计）
            if !level.is_empty() && first_file.is_none() && first_file_level.is_none() {
                first_file = Some(level[0].clone());
                first_file_level = Some(i);
            }
            contained_in += level.len();
            i += 1;
        });
        // 若重叠文件数 contained_in > 1（即一个键在多个文件中存在记录），调用 self.update_stats 方法更新统计信息（传入第一个重叠文件和其层级）
        if contained_in > 1 {
            // update_stats 方法会根据这些统计信息判断是否满足压缩条件（例如重叠文件过多影响查询效率），返回 true 表示需要触发新的压缩操作
            self.update_stats(GetStats {
                file: first_file,
                level: first_file_level.unwrap_or(0),
            })
        } else {
            // 若重叠文件数不多（contained_in <= 1），则返回 false，无需压缩
            false
        }
    }

    /// update_stats updates the number of seeks, and remembers files with too many seeks as
    /// compaction candidates. It returns true if a compaction makes sense.
    ///
    /// 这个 update_stats 方法是 LSM 树存储引擎中用于跟踪 SSTable 文件的查询频率并决定压缩压缩候选文件的工具函数。
    /// 它通过统计文件的查询次数（特别是 “seek 操作” 次数），判断那些访问过于频繁的文件，判断是否需要触发压缩（compaction）来优化查询性能。
    ///
    /// update_stats 的核心功能是实现 SSTable 文件的访问频率跟踪和压缩触发判断：
    ///
    /// 通过 allowed_seeks 衰减，量化文件的访问热度；
    ///
    /// 自动标记过热文件为压缩候选，避免频繁查询拖慢性能；
    ///
    /// 为 maybe_do_compaction 等压缩逻辑提供决策依据，确保存储系统长期高效。
    pub fn update_stats(&mut self, stats: GetStats) -> bool {
        // 处理该文件的统计信息
        if let Some(file) = stats.file {
            // allowed_seeks 是该文件允许的 “最大 seek 次数阈值”（初始值通常根据文件大小计算，越大的文件阈值越高）；
            // 当 allowed_seeks <= 1（即将耗尽）且当前没有其他压缩候选文件时：
            // 将该文件标记为 file_to_compact（待压缩文件）；
            // 记录其所在层级（file_to_compact_lvl）；
            // 返回 true，通知调用方需要触发压缩。
            if file.borrow().allowed_seeks <= 1 && self.file_to_compact.is_none() {
                self.file_to_compact = Some(file);  // 标记该文件为压缩候选
                self.file_to_compact_lvl = stats.level; // 记录该文件所在的层级
                return true; // 需要触发压缩
            } else if file.borrow().allowed_seeks > 0 {
                // 若文件仍有剩余的 allowed_seeks 次数，则每次查询命中后减 1；
                // 这是一种 “衰减机制”：访问越频繁的文件，allowed_seeks 消耗越快，越容易被标记为压缩候选。
                file.borrow_mut().allowed_seeks -= 1;  // 每次查询命中，减少一次允许的 seek 次数
            }
        }
        false
    }

    /// max_next_level_overlapping_bytes returns how many bytes of tables are overlapped in l+1 by
    /// tables in l, for the maximum case.
    fn max_next_level_overlapping_bytes(&self) -> usize {
        let mut max = 0;
        for lvl in 1..NUM_LEVELS - 1 {
            for f in &self.files[lvl] {
                let f = f.borrow();
                let ols = self.overlapping_inputs(lvl + 1, &f.smallest, &f.largest);
                let sum = total_size(ols.iter());
                if sum > max {
                    max = sum;
                }
            }
        }
        max
    }

    /// overlap_in_level returns true if the specified level's files overlap the range [smallest; largest].
    ///
    /// 这个 overlap_in_level 方法用于判断数据库指定层级（Level）的 SSTable 文件中，是否存在与目标键范围 [smallest, largest] 重叠的文件。
    /// 它是 LSM 树压缩策略中的关键判断工具，用于决定哪些层级需要参与压缩操作。
    ///
    /// 方法接收三个参数：level: usize（层级索引）、smallest: UserKey<'a>（目标范围的最小键）、largest: UserKey<'a>（目标范围的最大键），返回 bool（true 表示存在重叠文件，false 表示无）
    pub fn overlap_in_level<'a>(
        &self,
        level: usize,
        smallest: UserKey<'a>,
        largest: UserKey<'a>,
    ) -> bool {
        // 通过断言确保输入的层级是合法的（如 Level 0 到 Level 6，具体取决于 NUM_LEVELS 的定义），避免越界访问。
        assert!(level < NUM_LEVELS); // 确保层级索引在有效范围内（0 到 NUM_LEVELS-1）
        // Level 0 特性：该层级的 SSTable 文件之间可能存在键范围重叠（因直接由 immutable 内存表压缩生成，未经过范围排序），且文件本身的键范围可能不连续；
        // 判断逻辑：调用 some_file_overlaps_range_disjoint 检查是否有任何文件的键范围与 [smallest, largest] 重叠。该函数针对 “文件范围可能重叠且无序” 的场景设计，会逐个检查文件。
        if level == 0 {
            some_file_overlaps_range_disjoint(
                &InternalKeyCmp(self.user_cmp.clone()), // 内部键比较器
                &self.files[level],   // Level 0 的所有 SSTable 文件
                smallest, // 目标范围最小键
                largest,   // 目标范围最大键
            )
        } else {
            // Level 1+ 特性：这些层级的 SSTable 文件经过严格排序，键范围不重叠且连续（每个文件的键范围完全包含在其前后文件之间）；
            // 判断逻辑：调用 some_file_overlaps_range 检查是否有文件与目标范围重叠。该函数利用 “文件有序且不重叠” 的特性，可通过二分查找快速定位可能重叠的文件，无需遍历所有文件，效率更高。
            some_file_overlaps_range(
                &InternalKeyCmp(self.user_cmp.clone()),  // 内部键比较器
                &self.files[level],   // 当前层级的所有 SSTable 文件
                smallest,  // 目标范围最小键
                largest,   // 目标范围最大键
            )
        }
    }

    /// overlapping_inputs returns all files that may contain keys between begin and end.
    ///
    /// 这个 overlapping_inputs 方法是 LSM 树版本（Version）结构体中用于筛选指定层级（level）中，键范围与目标区间 [begin, end] 重叠的所有 SSTable 文件的核心逻辑。
    /// 它适配了 LSM 树不同层级（尤其是 Level 0）的文件特性，确保准确收集所有可能包含目标范围内键的文件，为压缩、查询等操作提供基础。
    ///
    /// Level 0	文件无序、可能互相重叠。找到重叠文件后，若其范围超出当前搜索区间，需扩展范围并重新遍历，避免遗漏其他重叠文件（因文件重叠可能导致 “跨范围” 关联）。
    ///
    /// Level 1+ 文件有序、键范围不重叠。一次遍历即可筛选所有重叠文件，无需扩展范围（文件有序性确保不会遗漏）。
    ///
    ///
    /// overlapping_inputs 的核心价值是精准收集目标范围内的所有相关 SSTable 文件，为后续操作提供基础：
    ///
    /// 1.支撑压缩策略
    ///
    /// 在 compact_range（手动压缩）等操作中，需先通过该方法找到与目标范围重叠的文件，才能生成压缩任务（合并这些文件、清理无效数据）。
    ///
    /// 2.优化查询效率
    ///
    /// 在查询某个键范围时，可通过该方法快速定位可能包含目标键的文件，避免遍历所有文件，减少 I/O 开销。
    ///
    /// 3.确保数据完整性
    ///
    /// 尤其是对 Level 0 的 “范围扩展 + 重新搜索” 逻辑，避免因文件重叠导致的漏选，确保后续操作（如压缩、查询）能覆盖所有相关数据。
    ///
    /// 示例：Level 0 的范围扩展过程
    /// 假设 Level 0 有 3 个重叠文件：
    ///
    /// 文件 A：[a, c]
    ///
    /// 文件 B：[b, d]
    ///
    /// 文件 C：[d, f]
    ///
    /// 初始搜索范围为 [b, c]：
    ///
    /// 第一次 do_search：找到文件 A（[a,c] 与 [b,c] 重叠），发现 A 的最小键 a < b，返回扩展范围 [a, c]；
    ///
    /// 第二次 do_search：用 [a,c] 搜索，找到文件 A 和 B（[b,d] 与 [a,c] 重叠），发现 B 的最大键 d > c，返回扩展范围 [a, d]；
    ///
    /// 第三次 do_search：用 [a,d] 搜索，找到文件 A、B、C（[d,f] 与 [a,d] 重叠），无范围扩展，返回最终集合 [A,B,C]。
    pub fn overlapping_inputs(
        &self,
        level: usize, // 目标层级
        begin: InternalKey<'_>, // 目标范围起始内部键
        end: InternalKey<'_>, // 目标范围结束内部键
    ) -> Vec<FileMetaHandle> {
        assert!(level < NUM_LEVELS); // 确保层级合法
        // LSM 树的 InternalKey 包含 “用户键 + 序列号 + 操作类型”，而文件的键范围本质是用户键的范围，因此先通过 parse_internal_key 剥离元数据，提取纯用户键 ubegin（起始）和 uend（结束）；
        // 用 Vec<u8> 存储用户键，支持后续动态调整范围。
        // 从内部键（含序列号+操作类型）中提取纯用户键，作为初始搜索范围
        let (mut ubegin, mut uend) = (
            parse_internal_key(begin).2.to_vec(),
            parse_internal_key(end).2.to_vec(),
        );

        // 这是针对 Level 0 特性的关键设计：Level 0 的 SSTable 文件可能无序且互相重叠，找到一个重叠文件后，其键范围可能超出当前搜索范围，需扩展范围后重新搜索，避免遗漏其他重叠文件；
        // 非 Level 0 层级（文件有序不重叠）不会触发范围扩展，一次 do_search 即可完成。
        loop {
            match do_search(self, level, ubegin, uend) {
                // 若返回新的范围（Some((newubegin, newuend))），更新范围并重新搜索
                DoSearchResult(Some((newubegin, newuend)), _) => {
                    ubegin = newubegin;
                    uend = newuend;
                }
                // 若返回 None，说明范围无需扩展，返回最终重叠文件集合
                DoSearchResult(None, result) => return result,
            }
        }

        // the actual search happens in this inner function. This is done to enhance the control
        // flow. It takes the smallest and largest user keys and returns a new pair of user keys if
        // the search range should be expanded, or a list of overlapping files.
        // do_search 是实际执行 “筛选重叠文件 + 处理 Level 0 范围扩展” 的逻辑，返回自定义枚举 DoSearchResult（包含 “是否需要扩展范围” 和 “当前找到的重叠文件”）。其流程如下：
        fn do_search(
            myself: &Version,
            level: usize,
            ubegin: Vec<u8>,
            uend: Vec<u8>,
        ) -> DoSearchResult {
            let mut inputs = vec![]; // 存储当前找到的重叠文件
            for f_ in myself.files[level].iter() {  // 遍历当前层级的所有 SSTable 文件
                let f = f_.borrow(); // 获取文件元数据（含最小/最大内部键）
                // 提取文件的最小/最大用户键（剥离内部键元数据）
                let (fsmallest, flargest) = (
                    parse_internal_key(&f.smallest).2,
                    parse_internal_key(&f.largest).2,
                );
                // Skip files that are not overlapping.
                // 步骤1：判断文件与当前搜索范围是否完全不重叠
                let ubegin_nonempty = !ubegin.is_empty();
                let uend_nonempty = !uend.is_empty();
                // 文件的最大用户键 < 搜索范围的起始键 → 文件在范围之前，不重叠
                let block_before_current = ubegin_nonempty && myself.user_cmp.cmp(flargest, &ubegin) == Ordering::Less;
                // 文件的最小用户键 > 搜索范围的结束键 → 文件在范围之后，不重叠
                let block_after_current = uend_nonempty && myself.user_cmp.cmp(fsmallest, &uend) == Ordering::Greater;
                // 反证法：不满足上述两种情况 → 存在重叠
                let is_overlapping = !(block_before_current || block_after_current);
                // 步骤2：处理重叠文件
                if is_overlapping {
                    inputs.push(f_.clone());  // 将重叠文件加入结果集
                    // In level 0, files may overlap each other. Check if the new file begins
                    // before ubegin or ends after uend, and expand the range, if so. Then, restart
                    // the search.
                    // 步骤3：Level 0 特殊处理：扩展搜索范围并重新搜索
                    if level == 0 {
                        // 若文件的最小键 < 当前搜索起始键 → 扩展起始键到文件最小键
                        if !ubegin.is_empty() && myself.user_cmp.cmp(fsmallest, &ubegin) == Ordering::Less {
                            return DoSearchResult(Some((fsmallest.to_vec(), uend)), inputs);
                        } else if !uend.is_empty() && myself.user_cmp.cmp(flargest, &uend) == Ordering::Greater {
                            // 若文件的最大键 > 当前搜索结束键 → 扩展结束键到文件最大键
                            return DoSearchResult(Some((ubegin, flargest.to_vec())), inputs);
                        }
                    }
                } else {
                    continue; // 不重叠，跳过当前文件
                }
            }
            DoSearchResult(None, inputs) // 遍历结束，无范围扩展需求，返回当前结果
        }
    }

    /// new_concat_iter returns an iterator that iterates over the files in a level. Note that this
    /// only really makes sense for levels > 0.
    fn new_concat_iter(&self, level: usize) -> VersionIter {
        new_version_iter(
            self.files[level].clone(),
            self.table_cache.clone(),
            self.user_cmp.clone(),
        )
    }

    /// new_iters returns a set of iterators that can be merged to yield all entries in this version.
    ///
    /// 这个 new_iters 方法是数据库版本（Version）结构体中用于创建当前版本所有 SSTable 文件的迭代器集合的核心函数。
    /// 它为每个层级的 SSTable 生成对应的迭代器，供合并迭代器（MergingIter）整合，最终实现全库数据的有序遍历。
    pub fn new_iters(&self) -> Result<Vec<Box<dyn LdbIterator>>> {
        // 创建一个向量 iters，用于存储所有 SSTable 的迭代器；
        // 类型 Box<dyn LdbIterator> 支持不同类型的 SSTable 迭代器（如单文件迭代器、多文件串联迭代器）。
        let mut iters: Vec<Box<dyn LdbIterator>> = vec![];
        // self.files[0] 存储 Level 0 的所有 SSTable 文件（f 是单个 SSTable 的元数据）；
        // Level 0 的 SSTable 特性：文件之间可能存在键范围重叠（因直接由 immutable 内存表压缩生成），因此每个文件需要单独的迭代器；
        // 操作逻辑：
        // 通过 table_cache（表缓存）获取 SSTable 文件的句柄（get_table(f.borrow().num)?），避免重复打开文件；
        // 为每个 SSTable 创建迭代器（iter()），包装后加入 iters。
        for f in &self.files[0] {
            iters.push(Box::new(
                self.table_cache
                    .borrow_mut()
                    .get_table(f.borrow().num)?
                    .iter(),
            ));
        }
        // NUM_LEVELS 是 LSM 树的总层级数（通常 Level 0 到 Level 6 等）；
        // Level 1 及以上的 SSTable 特性：同一层级的文件键范围不重叠且有序，因此可以将整个层级的文件合并为一个串联迭代器；
        // 操作逻辑：
        // 若层级 l 存在 SSTable（!self.files[l].is_empty()），则调用 self.new_concat_iter(l) 创建该层级的串联迭代器；
        // 串联迭代器内部会按键顺序遍历该层级的所有文件，对外表现为一个单一的迭代器；
        // 将串联迭代器加入 iters。
        for l in 1..NUM_LEVELS {
            if !self.files[l].is_empty() {
                iters.push(Box::new(self.new_concat_iter(l)));
            }
        }
        // 返回包含所有层级 SSTable 迭代器的集合，供 merge_iterators 方法进一步整合（与内存表迭代器合并）。
        Ok(iters)
    }
}

/// new_version_iter returns an iterator over the entries in the specified ordered list of table
/// files.
pub fn new_version_iter(
    files: Vec<FileMetaHandle>,
    cache: Shared<TableCache>,
    ucmp: Rc<Box<dyn Cmp>>,
) -> VersionIter {
    VersionIter {
        files,
        cache,
        cmp: InternalKeyCmp(ucmp),
        current: None,
        current_ix: 0,
    }
}

/// VersionIter iterates over the entries in an ordered list of table files (specifically, for
/// example, the tables in a level).
///
/// Note that VersionIter returns entries of type Deletion.
///
/// 这段代码定义了 VersionIter 结构体，它是 LSM 树存储引擎中用于遍历特定层级（level）内有序 SSTable 文件集合的迭代器，主要功能是统一访问一个层级中所有 SSTable 的键值对数据。
pub struct VersionIter {
    // NOTE: Maybe we need to change this to Rc to support modification of the file set after
    // creation of the iterator. Versions should be immutable, though.
    /// 存储一个层级中所有 SSTable 文件的元数据句柄（FileMetaHandle），这些文件按键范围有序排列（非 L0 层的 SSTable 通常键范围不重叠且有序）。
    files: Vec<FileMetaHandle>,
    /// 指向表缓存（TableCache）的共享引用，用于高效获取 SSTable 的迭代器（避免重复打开磁盘文件）。
    cache: Shared<TableCache>,
    /// 内部键比较器，用于确保迭代过程中键的排序一致性（考虑键本身和序列编号）。
    cmp: InternalKeyCmp,

    /// 当前正在遍历的 SSTable 的迭代器。
    current: Option<TableIterator>,
    /// 当前处理的 SSTable 在 files 列表中的索引。
    current_ix: usize,
}

impl LdbIterator for VersionIter {
    fn advance(&mut self) -> bool {
        println!("----------version advance----------");
        assert!(!self.files.is_empty());

        if let Some(ref mut t) = self.current {
            // table iter
            if t.advance() {
                return true;
            } else if self.current_ix >= self.files.len() - 1 {
                // Already on last table; can't advance further.
                return false;
            }

            // Load next table if current table is exhausted and we have more tables to go through.
            self.current_ix += 1;
        }

        // Initialize iterator or load next table.
        if let Ok(tbl) = self
            .cache
            .borrow_mut()
            .get_table(self.files[self.current_ix].borrow().num)
        {
            self.current = Some(tbl.iter());
        } else {
            return false;
        }
        self.advance()
    }
    fn current(&self) -> Option<(Bytes, Bytes)> {
        println!("----------version current----------");
        if let Some(ref t) = self.current {
            t.current()
        } else {
            None
        }
    }
    fn seek(&mut self, key: &[u8]) {
        println!("----------version seek----------");
        if let Some(ix) = find_file(&self.cmp, &self.files, key) {
            if let Ok(tbl) = self
                .cache
                .borrow_mut()
                .get_table(self.files[ix].borrow().num)
            {
                let mut iter = tbl.iter();
                iter.seek(key);
                if iter.valid() {
                    self.current_ix = ix;
                    self.current = Some(iter);
                    return;
                }
            }
        }
        self.reset();
    }
    fn reset(&mut self) {
        println!("----------version reset----------");
        self.current = None;
        self.current_ix = 0;
    }
    fn valid(&self) -> bool {
        println!("----------version valid----------");
        self.current.as_ref().map(|t| t.valid()).unwrap_or(false)
    }
    fn prev(&mut self) -> bool {
        println!("----------version prev----------");
        if let Some(ref mut t) = self.current {
            if t.prev() {
                return true;
            } else if self.current_ix > 0 {
                let f = &self.files[self.current_ix - 1];
                // Find previous table, seek to last entry.
                if let Ok(tbl) = self.cache.borrow_mut().get_table(f.borrow().num) {
                    let mut iter = tbl.iter();
                    iter.seek(&f.borrow().largest);
                    // The saved largest key must be in the table.
                    assert!(iter.valid());
                    self.current_ix -= 1;
                    *t = iter;
                    return true;
                }
            }
        }
        self.reset();
        false
    }
}

/// total_size returns the sum of sizes of the given files.
/// 这个 total_size 函数是一个工具函数，用于计算一组 SSTable 文件的总大小（以字节为单位）。
pub fn total_size<'a, I: Iterator<Item = &'a FileMetaHandle>>(files: I) -> usize {
    /// 使用 fold 方法对迭代器进行折叠操作，从初始值 0 开始
    /// 对每个文件元数据句柄 f：
    /// 通过 f.borrow() 获取到实际的文件元数据（FileMeta）
    /// 取出元数据中的 size 字段（单个文件的大小）
    /// 累加到总和中
    files.fold(0, |a, f| a + f.borrow().size)
}

/// key_is_after_file returns true if the given user key is larger than the largest key in f.
///
/// 这个 key_is_after_file 函数用于判断给定的用户键（key）是否大于某个 SSTable 文件（f）中存储的最大键。
/// 它是 LSM 树中判断键与文件范围关系的基础工具，主要用于确定键是否完全位于某文件的键范围之后（无重叠）。
///
/// 函数接收三个参数：cmp（内部键比较器）、key（待判断的用户键）、f（SSTable 文件的元数据句柄），返回 bool（true 表示用户键在文件最大键之后，false 表示否则）
///
/// 示例说明
/// 假设：
///
/// SSTable 文件 f 的最大内部键为 user_key_10\x01\x00...（用户键为 user_key_10）；
///
/// 待判断的用户键为 user_key_20。
///
/// 调用 key_is_after_file(cmp, "user_key_20", &f) 时：
///
/// 解析 f.largest 得到 ulargest = "user_key_10"；
///
/// 比较 user_key_20 与 user_key_10，结果为 Ordering::Greater；
///
/// 函数返回 true，表示 user_key_20 在文件 f 的最大键之后。
fn key_is_after_file(cmp: &InternalKeyCmp, key: UserKey<'_>, f: &FileMetaHandle) -> bool {
    // FileMetaHandle 是 SSTable 文件元数据的智能指针（通常是 Rc<RefCell<FileMetaData>>），通过 borrow() 获取元数据引用，包含文件的最大键（f.largest）等信息。
    let f = f.borrow(); // 解引用文件句柄，获取 SSTable 的元数据
    // f.largest 是文件存储的最大内部键（包含用户键 + 序列号 + 操作类型）；
    // parse_internal_key 函数解析内部键，返回一个元组 (操作类型, 序列号, 用户键)，这里取第三个元素 ulargest 即文件的最大用户键。
    let ulargest = parse_internal_key(&f.largest).2;     // 从文件的最大内部键中提取用户键（剥离序列号和操作类型）
    // 非空检查：!key.is_empty() 确保待判断的用户键不是空键（空键通常不参与正常比较）；
    // 键比较：cmp.cmp_inner(key, ulargest) 使用内部比较器按数据库定义的规则比较两个用户键；
    // 结果判断：若比较结果为 Ordering::Greater（用户键 > 文件最大用户键），则返回 true，表示该键在文件的键范围之后。
    !key.is_empty() && cmp.cmp_inner(key, ulargest) == Ordering::Greater     // 判断用户键非空，且大于文件的最大用户键
}

/// key_is_before_file returns true if the given user key is larger than the largest key in f.
///
/// 这个 key_is_before_file 函数用于判断给定的用户键（key）是否小于某个 SSTable 文件（f）中存储的最小键。
/// 它是 LSM 树中判断 “键与文件范围位置关系” 的核心基础工具，主要用于确定键是否完全位于某文件的键范围之前（无任何重叠）。
///
/// 函数接收三个参数：cmp（内部键比较器）、key（待判断的用户键）、f（SSTable 文件的元数据句柄），返回 bool（true 表示用户键在文件最小键之前，false 表示否则
///
/// 示例说明
/// 假设场景：
///
/// SSTable 文件 f 的最小内部键为 user_key_5\x01\x00...（解析后得到最小用户键 usmallest = "user_key_5"）；
///
/// 待判断的用户键为 user_key_3。
///
/// 调用 key_is_before_file(cmp, "user_key_3", &f) 时：
///
/// 解析文件元数据得到 usmallest = "user_key_5"；
///
/// 比较 user_key_3 与 user_key_5，结果为 Ordering::Less；
///
/// 函数返回 true，表示 user_key_3 位于文件 f 的键范围之前。
fn key_is_before_file(cmp: &InternalKeyCmp, key: UserKey<'_>, f: &FileMetaHandle) -> bool {
    // FileMetaHandle 是 SSTable 文件元数据的智能指针（通常为 Rc<RefCell<FileMetaData>>），通过 borrow() 安全获取元数据引用，元数据中包含文件的最小键（f.smallest）等关键信息。
    let f = f.borrow();  // 解引用文件句柄，获取 SSTable 的完整元数据
    // f.smallest 是文件存储的最小内部键（InternalKey），格式为 “用户键 + 8 字节元数据”（元数据含序列号和操作类型）；
    // parse_internal_key 函数专门解析内部键，返回元组 (操作类型, 序列号, 用户键)，这里取第三个元素 usmallest，即该文件存储的最小用户键（UserKey）。
    let usmallest = parse_internal_key(&f.smallest).2; // 从文件的最小“内部键”中提取纯用户键（剥离内部元数据）
    // 非空校验：!key.is_empty() 排除空键的无效情况（空键通常不参与正常的键范围比较）；
    // 统一规则比较：cmp.cmp_inner(key, usmallest) 使用数据库全局统一的比较器（InternalKeyCmp），按用户定义的排序规则（如字典序）比较待判断键与文件最小用户键；
    // 结果判定：若比较结果为 Ordering::Less（待判断键 < 文件最小用户键），则返回 true，表示该键完全位于文件的键范围之前。
    !key.is_empty() && cmp.cmp_inner(key, usmallest) == Ordering::Less    // 核心判断：键非空，且用户键 < 文件的最小用户键
}

/// find_file returns the index of the file in files that potentially contains the internal key
/// key. files must not overlap and be ordered ascendingly. If no file can contain the key, None is
/// returned.
///
/// 二分查找法, 找到含有某一个key的文件, 返回文件索引
///
/// 这个 find_file 函数用于在有序且不重叠的 SSTable 文件集合中，查找可能包含指定内部键（key）的文件索引。
/// 它主要针对 LSM 树中 Level 1 及以上层级的 SSTable（这些层级的文件按键范围有序排列且不重叠），通过二分查找高效定位目标文件。
///
/// 示例说明
/// 假设 Level 1 有 3 个 SSTable 文件，按键范围排序：
///
/// 文件 0：[a, c]
///
/// 文件 1：[d, f]
///
/// 文件 2：[g, i]
///
/// 查找目标键 e 时：
///
/// 初始 left=0，right=3，mid=1（文件 1 的最大键为 f）；
///
/// 比较 f 与 e → f > e，故 right=1；
///
/// 循环继续，left=0，right=1，mid=0（文件 0 的最大键为 c）；
///
/// 比较 c 与 e → c < e，故 left=1；
///
/// 循环结束，right=1，返回 Some(1)（文件 1 可能包含 e）。
///
fn find_file(
    cmp: &InternalKeyCmp, // 内部键比较器
    files: &[FileMetaHandle], // SSTable 文件集合
    key: InternalKey<'_>, // 待查找的内部键
) -> Option<usize> {
    // 采用左闭右开区间 [left, right) 作为查找范围，初始覆盖所有文件（0 到 files.len()）。
    let (mut left, mut right) = (0, files.len()); // 左边界=0，右边界=文件总数
    // 核心逻辑：利用文件按键范围有序的特性（每个文件的键范围完全小于后一个文件），通过比较中间文件的最大键与目标键，不断缩小查找范围：
    // 若中间文件的最大键 小于 目标键，说明目标键不可能在该文件或其左侧文件中，左边界右移（left = mid + 1）；
    // 若中间文件的最大键 大于等于 目标键，说明目标键可能在该文件或其左侧文件中，右边界左移（right = mid）。
    while left < right {
        let mid = (left + right) / 2;  // 计算中间位置
        // 比较中间文件的最大键与目标键
        if cmp.cmp(&files[mid].borrow().largest, key) == Ordering::Less {
            // 中间文件的最大键 < 目标键 → 目标文件在右侧，移动左边界
            left = mid + 1;
        } else {
            // 中间文件的最大键 ≥ 目标键 → 目标文件在左侧或当前位置，移动右边界
            right = mid;
        }
    }
    // 循环结束后，right 指向第一个最大键大于等于目标键的文件；
    // 若该索引有效（right < files.len()），则返回 Some(right)（该文件可能包含目标键）；
    // 否则返回 None（所有文件的最大键都小于目标键，无匹配文件）。
    if right < files.len() {
        Some(right)  // 找到可能包含目标键的文件，返回索引
    } else {
        None  // 未找到符合条件的文件
    }
}

/// some_file_overlaps_range_disjoint returns true if any of the given disjoint files (i.e. level > 1) contain keys in the range defined by the user keys [smallest; largest].
///
/// 这个 some_file_overlaps_range_disjoint 函数用于判断 “无序且可能重叠” 的 SSTable 文件集合中，是否存在与目标键范围 [smallest, largest] 重叠的文件。
/// 它主要针对 LSM 树中 Level 0 的 SSTable（因 Level 0 文件可能无序且范围重叠）设计，是判断层级与键范围是否重叠的核心工具之一。
///
///
fn some_file_overlaps_range_disjoint(
    cmp: &InternalKeyCmp, // 内部键比较器
    files: &[FileMetaHandle], // SSTable 文件集合
    smallest: UserKey<'_>, // 目标范围最小键
    largest: UserKey<'_>, // 目标范围最大键
) -> bool {
    // 将用户提供的最小键（smallest）转换为内部键格式（含最大序列号）
    // LSM 树的内部键包含用户键 + 序列号 + 操作类型，这里用 MAX_SEQUENCE_NUMBER（最大序列号）确保覆盖该键的所有历史版本；
    // 目的是构建一个能匹配该用户键所有可能版本的内部键，用于后续文件查找。
    let ikey = LookupKey::new(smallest, MAX_SEQUENCE_NUMBER);
    // find_file 作用：在文件集合中查找是否有文件的键范围包含 ikey（即该文件的最小键 ≤ ikey ≤ 最大键），返回该文件的索引 ix；
    // key_is_before_file 作用：判断 largest（目标范围最大键）是否在文件 files[ix] 的键范围之前（即 largest < 文件的最小键）；
    // 逻辑推导：
    // 若找到包含起始键的文件（Some(ix)），且 largest 不在该文件范围之前（!key_is_before_file），说明 largest ≥ 文件的最小键，结合起始键在文件范围内，可推断目标范围与该文件重叠；
    // 若未找到包含起始键的文件（None），则目标范围与所有文件都不重叠。

    // 在文件集合中查找包含该内部键的文件索引
    if let Some(ix) = find_file(cmp, files, ikey.internal_key()) {
        // 若找到，判断目标范围的最大键是否在该文件的键范围之后
        !key_is_before_file(cmp, largest, &files[ix])
    } else {
        // 若未找到包含起始键的文件，返回 false
        false
    }
}

/// some_file_overlaps_range returns true if any of the given possibly overlapping files contains keys in the range [smallest; largest].
///
/// 这个 some_file_overlaps_range 函数用于判断 “有序且不重叠” 的 SSTable 文件集合中，是否存在与目标键范围 [smallest, largest] 重叠的文件。
/// 它主要针对 LSM 树中 Level 1 及以上层级的 SSTable（这些层级的文件经过排序，键范围不重叠且连续）设计，是判断层级与键范围是否重叠的核心工具之一。
///
/// some_file_overlaps_range 的核心价值是高效判断有序不重叠文件集合与目标键范围的重叠关系，主要作用包括：
///
/// 适配 Level 1+ 的特性
///
/// Level 1 及以上的 SSTable 经过压缩排序，文件按键范围有序排列且互不重叠（每个文件的键范围完全包含在其前后文件之间）。这种特性使得通过简单遍历即可判断重叠，且无需复杂的查找逻辑。
///
/// 支撑压缩决策
///
/// 在 overlap_in_level 方法中被调用，用于判断 Level 1+ 是否与目标键范围重叠，为压缩策略（如 compact_range）提供依据 —— 只有存在重叠时，该层级才需要参与压缩。
///
/// 简洁准确的范围判断
///
/// 通过 “排除完全不重叠的情况” 来确定重叠，逻辑简单且无遗漏，确保不会误判或漏判重叠文件。
fn some_file_overlaps_range(
    cmp: &InternalKeyCmp, // 内部键比较器
    files: &[FileMetaHandle], // SSTable 文件集合
    smallest: UserKey<'_>, // 目标范围最小键
    largest: UserKey<'_>, // 目标范围最大键
) -> bool {
    // 核心逻辑：通过 “反证法” 判断重叠 —— 如果一个文件与目标范围不满足 “完全不重叠”，则说明两者存在重叠。
    // 两个辅助函数的作用：
    // key_is_after_file(cmp, smallest, f)：判断目标范围的最小键（smallest）是否在文件 f 的最大键之后（即 smallest > f.largest），此时目标范围完全在文件之后，无重叠；
    // key_is_before_file(cmp, largest, f)：判断目标范围的最大键（largest）是否在文件 f 的最小键之前（即 largest < f.smallest），此时目标范围完全在文件之前，无重叠。
    // 逻辑推导：
    // 若上述两个条件至少满足一个（||），则目标范围与文件 f 完全不重叠；
    // 若上述两个条件都不满足（!()），则目标范围与文件 f 存在重叠（目标范围的一部分或全部落在文件的键范围内）。
    for f in files {
        // 判断目标范围与当前文件是否不重叠
        if !(key_is_after_file(cmp, smallest, f) || key_is_before_file(cmp, largest, f)) {
            return true;  // 存在重叠，返回 true
        }
    }
    false  // 所有文件均不重叠，返回 false
}

#[cfg(test)]
pub mod testutil {
    use super::*;
    use crate::cache::Cache;
    use crate::cmp::DefaultCmp;
    use crate::env::Env;
    use crate::key_types::ValueType;
    use crate::options::{self, Options};
    use crate::table_builder::TableBuilder;
    use crate::table_cache::table_file_name;
    use crate::types::{share, FileMetaData, FileNum};

    use std::path::Path;

    pub fn new_file(
        num: u64,
        smallest: &[u8],
        smallestix: u64,
        largest: &[u8],
        largestix: u64,
    ) -> FileMetaHandle {
        share(FileMetaData {
            allowed_seeks: 10,
            size: 163840,
            num,
            smallest: LookupKey::new(smallest, smallestix)
                .internal_key()
                .to_vec()
                .into(),
            largest: LookupKey::new(largest, largestix)
                .internal_key()
                .to_vec()
                .into(),
        })
    }

    /// write_table creates a table with the given number and contents (must be sorted!) in the
    /// memenv. The sequence numbers given to keys start with startseq.
    pub fn write_table(
        me: &dyn Env,
        contents: &[(&[u8], &[u8], ValueType)],
        startseq: u64,
        num: FileNum,
    ) -> FileMetaHandle {
        let dst = me
            .open_writable_file(Path::new(&table_file_name("db", num)))
            .unwrap();
        let mut seq = startseq;
        let keys: Vec<Vec<u8>> = contents
            .iter()
            .map(|&(k, _, typ)| {
                seq += 1;
                LookupKey::new_full(k, seq - 1, typ).internal_key().to_vec()
            })
            .collect();

        let mut tbl = TableBuilder::new(options::for_test(), dst);
        for i in 0..contents.len() {
            tbl.add(&keys[i], contents[i].1).unwrap();
            seq += 1;
        }

        let f = new_file(
            num,
            contents[0].0,
            startseq,
            contents[contents.len() - 1].0,
            startseq + (contents.len() - 1) as u64,
        );
        f.borrow_mut().size = tbl.finish().unwrap();
        f
    }

    pub fn make_version() -> (Version, Options) {
        let opts = options::for_test();
        let env = opts.env.clone();

        // The different levels overlap in a sophisticated manner to be able to test compactions
        // and so on.
        // The sequence numbers are in "natural order", i.e. highest levels have lowest sequence
        // numbers.

        // Level 0 (overlapping)
        let f2: &[(&[u8], &[u8], ValueType)] = &[
            (b"aac", b"val1", ValueType::TypeDeletion),
            (b"aax", b"val2", ValueType::TypeValue),
            (b"aba", b"val3", ValueType::TypeValue),
            (b"bab", b"val4", ValueType::TypeValue),
            (b"bba", b"val5", ValueType::TypeValue),
        ];
        let t2 = write_table(env.as_ref().as_ref(), f2, 26, 2);
        let f1: &[(&[u8], &[u8], ValueType)] = &[
            (b"aaa", b"val1", ValueType::TypeValue),
            (b"aab", b"val2", ValueType::TypeValue),
            (b"aac", b"val3", ValueType::TypeValue),
            (b"aba", b"val4", ValueType::TypeValue),
        ];
        let t1 = write_table(env.as_ref().as_ref(), f1, 22, 1);
        // Level 1
        let f3: &[(&[u8], &[u8], ValueType)] = &[
            (b"aaa", b"val0", ValueType::TypeValue),
            (b"cab", b"val2", ValueType::TypeValue),
            (b"cba", b"val3", ValueType::TypeValue),
        ];
        let t3 = write_table(env.as_ref().as_ref(), f3, 19, 3);
        let f4: &[(&[u8], &[u8], ValueType)] = &[
            (b"daa", b"val1", ValueType::TypeValue),
            (b"dab", b"val2", ValueType::TypeValue),
            (b"dba", b"val3", ValueType::TypeValue),
        ];
        let t4 = write_table(env.as_ref().as_ref(), f4, 16, 4);
        let f5: &[(&[u8], &[u8], ValueType)] = &[
            (b"eaa", b"val1", ValueType::TypeValue),
            (b"eab", b"val2", ValueType::TypeValue),
            (b"fab", b"val3", ValueType::TypeValue),
        ];
        let t5 = write_table(env.as_ref().as_ref(), f5, 13, 5);
        // Level 2
        let f6: &[(&[u8], &[u8], ValueType)] = &[
            (b"cab", b"val1", ValueType::TypeValue),
            (b"fab", b"val2", ValueType::TypeValue),
            (b"fba", b"val3", ValueType::TypeValue),
        ];
        let t6 = write_table(env.as_ref().as_ref(), f6, 10, 6);
        let f7: &[(&[u8], &[u8], ValueType)] = &[
            (b"gaa", b"val1", ValueType::TypeValue),
            (b"gab", b"val2", ValueType::TypeValue),
            (b"gba", b"val3", ValueType::TypeValue),
            (b"gca", b"val4", ValueType::TypeDeletion),
            (b"gda", b"val5", ValueType::TypeValue),
        ];
        let t7 = write_table(env.as_ref().as_ref(), f7, 5, 7);
        // Level 3 (2 * 2 entries, for iterator behavior).
        let f8: &[(&[u8], &[u8], ValueType)] = &[
            (b"haa", b"val1", ValueType::TypeValue),
            (b"hba", b"val2", ValueType::TypeValue),
        ];
        let t8 = write_table(env.as_ref().as_ref(), f8, 3, 8);
        let f9: &[(&[u8], &[u8], ValueType)] = &[
            (b"iaa", b"val1", ValueType::TypeValue),
            (b"iba", b"val2", ValueType::TypeValue),
        ];
        let t9 = write_table(env.as_ref().as_ref(), f9, 1, 9);

        let cache = TableCache::new("db", opts.clone(), share(Cache::new(128)), 100);
        let mut v = Version::new(share(cache), Rc::new(Box::new(DefaultCmp)));
        v.files[0] = vec![t1, t2];
        v.files[1] = vec![t3, t4, t5];
        v.files[2] = vec![t6, t7];
        v.files[3] = vec![t8, t9];
        (v, opts)
    }
}

#[cfg(test)]
mod tests {
    use super::testutil::*;
    use super::*;

    use crate::cmp::DefaultCmp;
    use crate::error::Result;
    use crate::merging_iter::MergingIter;
    use crate::options;
    use crate::test_util::{test_iterator_properties, LdbIteratorIter};

    #[test]
    fn test_version_concat_iter() {
        let v = make_version().0;

        let expected_entries = [0, 9, 8, 4];
        (1..4).for_each(|l| {
            let mut iter = v.new_concat_iter(l);
            let iter = LdbIteratorIter::wrap(&mut iter);
            assert_eq!(iter.count(), expected_entries[l]);
        });
    }

    #[test]
    fn test_version_concat_iter_properties() {
        let v = make_version().0;
        let iter = v.new_concat_iter(3);
        test_iterator_properties(iter);
    }

    #[test]
    fn test_version_max_next_level_overlapping() {
        let v = make_version().0;
        assert_eq!(218, v.max_next_level_overlapping_bytes());
    }

    #[test]
    fn test_version_all_iters() {
        let v = make_version().0;
        let iters = v.new_iters().unwrap();
        let mut opt = options::for_test();
        opt.cmp = Rc::new(Box::new(InternalKeyCmp(Rc::new(Box::new(DefaultCmp)))));

        let mut miter = MergingIter::new(opt.cmp.clone(), iters);
        assert_eq!(LdbIteratorIter::wrap(&mut miter).count(), 30);

        // Check that all elements are in order.
        let init = LookupKey::new(b"000", MAX_SEQUENCE_NUMBER);
        let cmp = InternalKeyCmp(Rc::new(Box::new(DefaultCmp)));
        LdbIteratorIter::wrap(&mut miter).fold(init.internal_key().to_vec(), |b, (k, _)| {
            assert!(cmp.cmp(&b, &k) == Ordering::Less);
            k
        });
    }

    #[test]
    fn test_version_summary() {
        let v = make_version().0;
        let expected = "level 0: 2 files, 483 bytes ([(1, 232), (2, 251)]); level 1: 3 files, 651 \
                        bytes ([(3, 218), (4, 216), (5, 217)]); level 2: 2 files, 468 bytes ([(6, \
                        218), (7, 250)]); level 3: 2 files, 400 bytes ([(8, 200), (9, 200)]); ";
        assert_eq!(expected, &v.level_summary());
    }

    #[test]
    fn test_version_get_simple() {
        let v = make_version().0;
        let cases: &[(&[u8], u64, Result<Option<Vec<u8>>>)] = &[
            (b"aaa", 1, Ok(None)),
            (b"aaa", 100, Ok(Some("val1".as_bytes().to_vec()))),
            (b"aaa", 21, Ok(Some("val0".as_bytes().to_vec()))),
            (b"aab", 0, Ok(None)),
            (b"aab", 100, Ok(Some("val2".as_bytes().to_vec()))),
            (b"aac", 100, Ok(None)),
            (b"aac", 25, Ok(Some("val3".as_bytes().to_vec()))),
            (b"aba", 100, Ok(Some("val3".as_bytes().to_vec()))),
            (b"aba", 25, Ok(Some("val4".as_bytes().to_vec()))),
            (b"daa", 100, Ok(Some("val1".as_bytes().to_vec()))),
            (b"dab", 1, Ok(None)),
            (b"dac", 100, Ok(None)),
            (b"gba", 100, Ok(Some("val3".as_bytes().to_vec()))),
            // deleted key
            (b"gca", 100, Ok(None)),
            (b"gbb", 100, Ok(None)),
        ];

        cases
            .iter()
            .for_each(|c| match v.get(LookupKey::new(c.0, c.1).internal_key()) {
                Ok(Some((val, _))) => assert_eq!(c.2.as_ref().unwrap().as_ref().unwrap(), &val),
                Ok(None) => assert!(c.2.as_ref().unwrap().as_ref().is_none()),
                Err(_) => assert!(c.2.is_err()),
            });
    }

    #[test]
    fn test_version_get_overlapping_basic() {
        let v = make_version().0;

        // Overlapped by tables 1 and 2.
        let ol = v.get_overlapping(LookupKey::new(b"aay", 50).internal_key());
        // Check that sorting order is newest-first in L0.
        assert_eq!(2, ol[0][0].borrow().num);
        // Check that table from L1 matches.
        assert_eq!(3, ol[1][0].borrow().num);

        let ol = v.get_overlapping(LookupKey::new(b"cb", 50).internal_key());
        assert_eq!(3, ol[1][0].borrow().num);
        assert_eq!(6, ol[2][0].borrow().num);

        let ol = v.get_overlapping(LookupKey::new(b"x", 50).internal_key());
        (0..NUM_LEVELS).for_each(|i| {
            assert!(ol[i].is_empty());
        });
    }

    #[test]
    fn test_version_overlap_in_level() {
        let v = make_version().0;

        for &(level, (k1, k2), want) in &[
            (0, ("000".as_bytes(), "003".as_bytes()), false),
            (0, ("aa0".as_bytes(), "abx".as_bytes()), true),
            (1, ("012".as_bytes(), "013".as_bytes()), false),
            (1, ("abc".as_bytes(), "def".as_bytes()), true),
            (2, ("xxx".as_bytes(), "xyz".as_bytes()), false),
            (2, ("gac".as_bytes(), "gaz".as_bytes()), true),
        ] {
            if want {
                assert!(v.overlap_in_level(level, k1, k2));
            } else {
                assert!(!v.overlap_in_level(level, k1, k2));
            }
        }
    }

    #[test]
    fn test_version_pick_memtable_output_level() {
        let v = make_version().0;

        for c in [
            ("000".as_bytes(), "abc".as_bytes(), 0),
            ("gab".as_bytes(), "hhh".as_bytes(), 1),
            ("000".as_bytes(), "111".as_bytes(), 2),
        ]
        .iter()
        {
            assert_eq!(c.2, v.pick_memtable_output_level(c.0, c.1));
        }
    }

    #[test]
    fn test_version_overlapping_inputs() {
        let v = make_version().0;

        time_test!("overlapping-inputs");
        {
            // Range is expanded in overlapping level-0 files.
            let from = LookupKey::new("aab".as_bytes(), MAX_SEQUENCE_NUMBER);
            let to = LookupKey::new("aae".as_bytes(), 0);
            let r = v.overlapping_inputs(0, from.internal_key(), to.internal_key());
            assert_eq!(r.len(), 2);
            assert_eq!(r[0].borrow().num, 1);
            assert_eq!(r[1].borrow().num, 2);
        }
        {
            let from = LookupKey::new("cab".as_bytes(), MAX_SEQUENCE_NUMBER);
            let to = LookupKey::new("cbx".as_bytes(), 0);
            // expect one file.
            let r = v.overlapping_inputs(1, from.internal_key(), to.internal_key());
            assert_eq!(r.len(), 1);
            assert_eq!(r[0].borrow().num, 3);
        }
        {
            let from = LookupKey::new("cab".as_bytes(), MAX_SEQUENCE_NUMBER);
            let to = LookupKey::new("ebx".as_bytes(), 0);
            let r = v.overlapping_inputs(1, from.internal_key(), to.internal_key());
            // Assert that correct number of files and correct files were returned.
            assert_eq!(r.len(), 3);
            assert_eq!(r[0].borrow().num, 3);
            assert_eq!(r[1].borrow().num, 4);
            assert_eq!(r[2].borrow().num, 5);
        }
        {
            let from = LookupKey::new("hhh".as_bytes(), MAX_SEQUENCE_NUMBER);
            let to = LookupKey::new("ijk".as_bytes(), 0);
            let r = v.overlapping_inputs(2, from.internal_key(), to.internal_key());
            assert_eq!(r.len(), 0);
            let r = v.overlapping_inputs(1, from.internal_key(), to.internal_key());
            assert_eq!(r.len(), 0);
        }
    }

    #[test]
    fn test_version_record_read_sample() {
        let mut v = make_version().0;
        let k = LookupKey::new("aab".as_bytes(), MAX_SEQUENCE_NUMBER);
        let only_in_one = LookupKey::new("cax".as_bytes(), MAX_SEQUENCE_NUMBER);

        assert!(!v.record_read_sample(k.internal_key()));
        assert!(!v.record_read_sample(only_in_one.internal_key()));

        for fs in v.files.iter() {
            for f in fs {
                f.borrow_mut().allowed_seeks = 0;
            }
        }
        assert!(v.record_read_sample(k.internal_key()));
    }

    #[test]
    fn test_version_key_ordering() {
        time_test!();
        let fmh = new_file(1, &[1, 0, 0], 0, &[2, 0, 0], 1);
        let cmp = InternalKeyCmp(Rc::new(Box::new(DefaultCmp)));

        // Keys before file.
        for k in &[&[0][..], &[1], &[1, 0], &[0, 9, 9, 9]] {
            assert!(key_is_before_file(&cmp, k, &fmh));
            assert!(!key_is_after_file(&cmp, k, &fmh));
        }
        // Keys in file.
        for k in &[
            &[1, 0, 0][..],
            &[1, 0, 1],
            &[1, 2, 3, 4],
            &[1, 9, 9],
            &[2, 0, 0],
        ] {
            assert!(!key_is_before_file(&cmp, k, &fmh));
            assert!(!key_is_after_file(&cmp, k, &fmh));
        }
        // Keys after file.
        for k in &[&[2, 0, 1][..], &[9, 9, 9], &[9, 9, 9, 9]] {
            assert!(!key_is_before_file(&cmp, k, &fmh));
            assert!(key_is_after_file(&cmp, k, &fmh));
        }
    }

    #[test]
    fn test_version_file_overlaps() {
        time_test!();

        let files_disjoint = [
            new_file(1, &[2, 0, 0], 0, &[3, 0, 0], 1),
            new_file(2, &[3, 0, 1], 0, &[4, 0, 0], 1),
            new_file(3, &[4, 0, 1], 0, &[5, 0, 0], 1),
        ];
        let files_joint = [
            new_file(1, &[2, 0, 0], 0, &[3, 0, 0], 1),
            new_file(2, &[2, 5, 0], 0, &[4, 0, 0], 1),
            new_file(3, &[3, 5, 1], 0, &[5, 0, 0], 1),
        ];
        let cmp = InternalKeyCmp(Rc::new(Box::new(DefaultCmp)));

        assert!(some_file_overlaps_range(
            &cmp,
            &files_joint,
            &[2, 5, 0],
            &[3, 1, 0]
        ));
        assert!(some_file_overlaps_range(
            &cmp,
            &files_joint,
            &[2, 5, 0],
            &[7, 0, 0]
        ));
        assert!(some_file_overlaps_range(
            &cmp,
            &files_joint,
            &[0, 0],
            &[2, 0, 0]
        ));
        assert!(some_file_overlaps_range(
            &cmp,
            &files_joint,
            &[0, 0],
            &[7, 0, 0]
        ));
        assert!(!some_file_overlaps_range(
            &cmp,
            &files_joint,
            &[0, 0],
            &[0, 5]
        ));
        assert!(!some_file_overlaps_range(
            &cmp,
            &files_joint,
            &[6, 0],
            &[7, 5]
        ));

        assert!(some_file_overlaps_range_disjoint(
            &cmp,
            &files_disjoint,
            &[2, 0, 1],
            &[2, 5, 0]
        ));
        assert!(some_file_overlaps_range_disjoint(
            &cmp,
            &files_disjoint,
            &[3, 0, 1],
            &[4, 9, 0]
        ));
        assert!(some_file_overlaps_range_disjoint(
            &cmp,
            &files_disjoint,
            &[2, 0, 1],
            &[6, 5, 0]
        ));
        assert!(some_file_overlaps_range_disjoint(
            &cmp,
            &files_disjoint,
            &[0, 0, 1],
            &[2, 5, 0]
        ));
        assert!(some_file_overlaps_range_disjoint(
            &cmp,
            &files_disjoint,
            &[0, 0, 1],
            &[6, 5, 0]
        ));
        assert!(!some_file_overlaps_range_disjoint(
            &cmp,
            &files_disjoint,
            &[0, 0, 1],
            &[0, 1]
        ));
        assert!(!some_file_overlaps_range_disjoint(
            &cmp,
            &files_disjoint,
            &[6, 0, 1],
            &[7, 0, 1]
        ));
    }

    #[test]
    fn test_vec(){
        let mut vec = vec![1];
        vec.reserve(10);
        println!("{}", vec.capacity());
        assert!(vec.capacity() >= 11);
    }
}
